package plgbase

import (
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/utils"
	"fmt"
	"gitee.com/ymofen/gobase/gotask"
	"gitee.com/ymofen/gobase/gotask/grouptask"
	"gitee.com/ymofen/gobase/subpub"
	"gitee.com/ymofen/supervalue"
	"time"
)

type PlgSubPubNode struct {
	PlgBase
	// 0:都显示, 1:显示>=1的消息
	msgShowLvl int8
	msglst     *utils.SyncCycleList

	subMonitor *utils.MonitorSpeedSizeRec
	subNode    *BaseSubNode

	pubMonitor *utils.MonitorSpeedSizeRec
	pubNode    *BasePubNode

	lastmsg string
}

func (this *PlgSubPubNode) onSubBuf(buf []byte) {
	this.subMonitor.DeltaBuf(buf)
	this.pubNode.Pub("", 0, buf)
}

func (this *PlgSubPubNode) onRecvSub(id, topic string, args ...interface{}) (ok bool) {
	if len(args) == 0 {
		return false
	}
	action, _ := args[0].(int)
	if action == subpub.ACTION_RECV_TOPIC_DATA {
		//recvTopic := args[1].(string)
		recvData := args[2]
		if buf, ok := recvData.([]byte); ok {
			this.onSubBuf(buf)
		} else if str, ok := recvData.(string); ok {
			this.onSubBuf([]byte(str))
		} else {
			return false
		}
		return true
	} else if action == subpub.ACTION_SERVER_CONN_RECV_BUF {
		// action, sessionid, buf
		recvData := args[2]
		if buf, ok := recvData.([]byte); ok {
			this.onSubBuf(buf)
		} else if str, ok := recvData.(string); ok {
			this.onSubBuf([]byte(str))
		} else {
			return false
		}
		return true
	} else if action == subpub.ACTION_RECVBUF {
		recvData := args[1]
		if buf, ok := recvData.([]byte); ok {
			this.onSubBuf(buf)
		} else if str, ok := recvData.(string); ok {
			this.onSubBuf([]byte(str))
		} else {
			return false
		}

		return true
	} else if action == subpub.ACTION_CONNECTED {

	} else if action == subpub.ACTION_DISCONNECTED {

	} else if action == subpub.ACTION_CONNECTFAIL {
		err := args[1].(error)
		this.lastmsg = fmt.Sprintf("[%s]sub连接出现异常:%s", utils.NowString(), err.Error())
		this.LogWarnMessage(2, "sub连接出现异常:%s", err.Error())
	} else if action == subpub.ACTION_ERROR {
		if err, ok := args[1].(error); ok {
			this.lastmsg = fmt.Sprintf("[%s]sub异常:%s", utils.NowString(), err.Error())
			if len(args) > 2 {
				this.LogWarnMessage(2, "sub出现异常:%s, %s", err.Error(), fmt.Sprint(args[2:]))
			} else {
				this.LogWarnMessage(2, "sub出现异常:%s", err.Error())
			}
		} else {
			this.LogWarnMessage(2, "sub出现异常:%s", fmt.Sprint(args[1:]))
		}
	} else if action == subpub.ACTION_MESSAGE {
		this.lastmsg = fmt.Sprintf("[%s]sub消息:%s", utils.NowString(), fmt.Sprint(args[1:]))
		this.LogInfoMessage(2, "[sub]%s", fmt.Sprint(args[1:]))
	} else {
		//str := fmt.Sprint(args...)
		//this.LogInfoMessage(0, "%s", str)

	}
	return true
}

func (this *PlgSubPubNode) onRecvPub(id, topic string, args ...interface{}) (ok bool) {
	if len(args) == 0 {
		return false
	}
	action, _ := args[0].(int)
	if action == subpub.ACTION_SEND_BUF { // caster 可能多并发
		buf := args[1].([]byte)
		l := int64(len(buf))
		this.pubMonitor.Delta(l)
		return true
	} else if action == subpub.ACTION_CONNECTED {
		this.LogInfoMessage(2, "pub连接成功")
	} else if action == subpub.ACTION_DISCONNECTED {
		this.LogWarnMessage(2, "pub连接断开:%s", args[1].(string))
	} else if action == subpub.ACTION_CONNECTFAIL {
		err := args[1].(error)
		this.lastmsg = fmt.Sprintf("[%s]pub连接出现异常:%s", utils.NowString(), err.Error())
		this.LogWarnMessage(2, "pub连接出现异常:%s", err.Error())
	} else if action == subpub.ACTION_ERROR {
		if err, ok := args[1].(error); ok {
			this.lastmsg = fmt.Sprintf("[%s]pub异常:%s", utils.NowString(), err.Error())
			this.LogWarnMessage(2, "pub出现异常:%s", err.Error())
		}
	} else if action == subpub.ACTION_MESSAGE {
		this.lastmsg = fmt.Sprintf("[%s]消息:%s", utils.NowString(), fmt.Sprint(args[1:]))
		this.LogInfoMessage(2, "[pub]%s", fmt.Sprint(args[1:]))
	} else {
		//str := fmt.Sprint(args...)
		//this.LogInfoMessage(0, "%s", str)

	}
	return true
}

//{
//"id": "xxx",
//"type": "plgs.subpubnode",
//"conf": {
//	"subconf":"conntype=tarmserial&connstr=119.96.169.117:6379&redistopic=",
//	"pubconf":"conntype=tcpclient&connstr=119.96.169.117:6379",
//}
//}

func (this *PlgSubPubNode) ReloadConf(conf *wrapper.SuperValue) error {
	if conf == nil {
		return nil
	}

	if this.msglst == nil {
		this.msglst = utils.NewSyncCycleList()
		this.msgShowLvl = 2
	}

	this.msgShowLvl = conf.Int8ByName("msgshowlvl", this.msgShowLvl)

	if this.subMonitor == nil {
		this.subMonitor = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitor(this.subMonitor)
	}

	if this.pubMonitor == nil {
		this.pubMonitor = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitor(this.pubMonitor)
	}

	if this.pubNode == nil {
		this.pubNode = NewBasePubNode(this.PlugId)
	}
	this.pubNode.ReloadConf(conf.ValueByName("pubconf"), this.onRecvPub)

	if this.subNode == nil {
		this.subNode = NewBaseSubNode(this.PlugId)
	}
	this.subNode.ReloadConf(conf.ValueByName("subconf"), this.onRecvSub)

	gotask.DefaultPatrolTask().AddTask(this.PlugId, time.Second*conf.DurationByName("status-interval-secs", 30), func(id interface{}, args ...interface{}) {
		grouptask.DefaultGroupTask().PostTaskFunc(this, func() {

		})
	})
	return nil
}

func (this *PlgSubPubNode) Close() (err error) {

	this.subNode.Close()
	this.pubNode.Close()
	this.PlgBase.Close()
	gotask.DefaultPatrolTask().DelTask(this.PlugId)
	utils.Monitor.RemoveMonitorRec(this.subMonitor)
	utils.Monitor.RemoveMonitorRec(this.pubMonitor)
	return
}

func (this *PlgSubPubNode) DoStatusRequestRequest(args ...interface{}) (resp interface{}) {
	reqArgs, _ := NewSVArgs(true, args...)
	if reqArgs != nil {
		defer reqArgs.Clear()
		itype := reqArgs.IntByName("type", 0)
		if itype == 99 {
			json := wrapper.NewSVObjectCache(true)
			json.SetKeyValue("submonitor", this.subMonitor.Info())
			json.SetKeyValue("subnode", this.subNode.StatusText())
			json.SetKeyValue("pubmonitor", this.pubMonitor.Info())
			json.SetKeyValue("pubnode", this.pubNode.StatusText())
			json.SetKeyValue("lastmsg", this.lastmsg)
			json.SetKeyValue("msgShowLvl", this.msgShowLvl)
			json.SetKeyValues("now", utils.NowString())
			return json
		} else if itype == 98 {
			return this.msglst
		} else if itype == 97 {
			lvl := int8(reqArgs.IntByName("lvl", -1))
			if lvl != -1 {
				this.msgShowLvl = lvl
			}
			return utils.BuildJSONString("code", 0, "lvl", this.msgShowLvl)
		} else if itype == 1 {
			json := wrapper.NewSVArrayCache(true)
			//searchval := reqArgs.StringByName("searchval", "")
			//sn := 0
			//maxn := reqArgs.IntByName("n", 100)
			//this.lst.ListRangeV2(func(key, val interface{}) bool {
			//	pojo := val.(*VrsSitePojo)
			//	ok := true
			//	if len(searchval) > 0 {
			//		ok = strings.Contains(pojo.id, searchval)
			//	}
			//	return ok
			//}, func(k0, v0, k1, v1 interface{}) bool {
			//	p0 := v0.(*VrsSitePojo)
			//	p1 := v1.(*VrsSitePojo)
			//	return p0.id < p1.id
			//}, func(key, val interface{}) bool {
			//	sn++
			//	pojo := val.(*VrsSitePojo)
			//	itm := json.AppendObject()
			//	itm.SetKeyValue("sn", sn)
			//	itm.SetKeyValue("id", pojo.id)
			//	return sn < maxn
			//})

			return json
		} else {
			//resp = this.DoRequest(args...)
			//if resp != nil {
			//	return resp
			//}
		}
	}
	var sb utils.BytesBuilder
	PrjHttpUtils.AppendIndexBody(&sb, "status", "status?type=99")
	PrjHttpUtils.AppendIndexBody(&sb, "msglist", "status?type=98&n=100&searchval=")
	PrjHttpUtils.AppendIndexBody(&sb, "msgshowlvl", "status?type=97&lvl=")
	PrjHttpUtils.AppendIndexBody(&sb, "lst", "status?type=1&n=100&searchval=")
	return sb.Bytes()
}

/*
msglvl: 值越大信息约重要
*/
func (this *PlgSubPubNode) LogInfoMessage(msglvl int8, s string, args ...interface{}) {
	if msglvl < this.msgShowLvl {
		return
	}
	this.msglst.LogMsg(s, args...)

	utils.Log.Msgf(this.PlugId, utils.LogLV_INFO, 1, s, args...)
}

func (this *PlgSubPubNode) LogWarnMessage(msglvl int8, s string, args ...interface{}) {
	if msglvl < this.msgShowLvl {
		return
	}
	this.msglst.LogMsg(s, args...)

	utils.Log.Msgf(this.PlugId, utils.LogLV_WARN, 1, s, args...)
}

func CreatePlgSubPubNodeFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &PlgSubPubNode{}

	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}

func init() {
	utils.RegisterCreate("plgs.subpubnode", CreatePlgSubPubNodeFunc)
}
